/**
 * Copyright (c) 2013-2019 Nikita Koksharov
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson.pubsub;

import java.util.Iterator;
import java.util.LinkedHashSet;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * (以PublishSubscribe中用到AsyncSemaphore进行讲解。其他情况类推即可)
 * 用于限制当前服务器上'尝试与redis建立订阅关系的线程数量'。
 * 由于PublishSubscribe会初始化50个信号量为1的AsyncSemaphore，因此同一时刻，当前服务器上最多有50个线程同时向redis发送SUB命令。
 */
public class AsyncSemaphore {
    /**
     * 自定义的static子类Entry。
     * 由于AsyncSemaphore被初始化为信号量1，所以只要AsyncSemaphore被线程A获取且正在发送SUB命令，
     * 那么继续获取该AsyncSemaphore的线程B必然无法继续发送SUB命令，需要等待线程A发送完毕。
     * 这里的子类Entry就是用来记录线程B的。
     */
    private static class Entry {
        /**
         * 属性1:线程B想干嘛。其实就是发送SUB命令的Runnable任务
         */
        private Runnable runnable;
        /**
         * 属性2:线程B需要的信号量。该值总会是1
         */
        private int permits;

        Entry(Runnable runnable, int permits) {
            super();
            this.runnable = runnable;
            this.permits = permits;
        }

        public int getPermits() {
            return permits;
        }

        public Runnable getRunnable() {
            return runnable;
        }

        /**
         * 相同异步任务的Entry将获得相同的hashcode。
         */
        @Override
        @SuppressWarnings("AvoidInlineConditionals")
        public int hashCode() {
            final int prime = 31;
            int result = 1;
            result = prime * result + ((runnable == null) ? 0 : runnable.hashCode());
            return result;
        }

        /**
         * 只要异步任务相同，两个Entry就是equals的，不在乎permits是否相同
         */
        @Override
        public boolean equals(Object obj) {
            if (this == obj)
                return true;
            if (obj == null)
                return false;
            if (getClass() != obj.getClass())
                return false;
            Entry other = (Entry) obj;
            if (runnable == null) {
                if (other.runnable != null)
                    return false;
            } else if (!runnable.equals(other.runnable))
                return false;
            return true;
        }
    }



    /**
     * 属性1:原子int，代表这个AsyncSemaphore的总信号量。
     * 该值总会是1，代表同一时刻最多进入一个线程
     */
    private volatile int counter;
    /**
     * 属性2:final集合，保存上面定义的Entry子类。
     * 就是其他想要发送SUB命令的线程呗，被阻塞了。
     */
    private final Set<Entry> listeners = new LinkedHashSet<Entry>();

    /**
     * 构造函数
     *
     * @param permits AsyncSemaphore的总信号量，将赋给counter。该值总会是1
     */
    public AsyncSemaphore(int permits) {
        counter = permits;
    }

    /**
     * 尝试在指定时间内从AsyncSemaphore的总信号量中获取一个信号量
     *
     * @param timeoutMillis 指定时间的毫秒形式
     * @return 是否成功
     */
    public boolean tryAcquire(long timeoutMillis) {
        // 定义一个线程计数器latch，初始值就是1
        final CountDownLatch latch = new CountDownLatch(1);

        // 定义一个异步任务listener，任务内容就是将线程计数器latch - 1
        final Runnable listener = new Runnable() {
            @Override
            public void run() {
                latch.countDown();
            }
        };

        // 调用下方的acquire()方法，尝试从AsyncSemaphore的总信号量-1，如果成功，则执行上面定义的异步任务listener
        acquire(listener);
        // 如果能成功获取到 AsyncSemaphore 的信号量，那么listener异步任务就会被执行，于是latch的值就会变成0

        try {
            // await: 等待latch的值变成0且最多等待timeoutMillis毫秒
            boolean res = latch.await(timeoutMillis, TimeUnit.MILLISECONDS);
            // 显然如果上面能获取到AsyncSemaphore 的信号量，那么latch的值就是0，res就是true。超时了还不是0，res就是false
            if (!res) {
                // 没有获取到信号量且超时后，从AsyncSemaphore的listeners中删除上面定义的异步任务
                if (!remove(listener)) {
                    // 进入if说明listeners中已经不存在listener异步任务了
                    // 这是因为，可能在!res时listener还没有执行，但是刚进入if(!res)，由于其他线程执行release()，导致该线程的listeners也得到了执行
                    // 那么这里也执行release()
                    release();
                }
            }
            return res;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            if (!remove(listener)) {
                release();
            }
            return false;
        }
    }

    public int queueSize() {
        synchronized (this) {
            return listeners.size();
        }
    }

    public void removeListeners() {
        synchronized (this) {
            listeners.clear();
        }
    }

    /**
     * 一个线程尝试向Redis发送SUB命令，检查AsyncSemaphore的信号量是否空闲。
     * 方法非阻塞。空闲就发送，不空闲就入队列等待。
     *
     * @param listener 如果成功，就执行这个异步任务(即向Redis发送SUB命令)
     */
    public void acquire(Runnable listener) {
        acquire(listener, 1);
    }

    /**
     * 一个线程尝试向Redis发送SUB命令，检查AsyncSemaphore的信号量是否空闲。
     * 方法非阻塞。空闲就发送，不空闲就入队列等待。
     *
     * @param listener 如果成功，就执行这个异步任务(即向Redis发送SUB命令)
     * @param permits 想要获取的permits(此情景下总是1)
     */
    public void acquire(Runnable listener, int permits) {
        // 记录AsyncSemaphore的信号量是否空闲
        boolean run = false;

        // synchronized块。保证在当前服务器上总能串行运行。保证AsyncSemaphore的原子性
        synchronized (this) {
            if (counter < permits) {
                // 如果AsyncSemaphore的总信号量小于想要获取的permits，那显然是无法获取成功的
                // 于是将这个异步任务和它想要获取的permits制作成Entry，保存到listeners集合中
                listeners.add(new Entry(listener, permits));
                // 直接返回
                return;
            } else {
                // 否则就从AsyncSemaphore的总信号量中减去permits
                counter -= permits;
                // 记录AsyncSemaphore的信号量空闲
                run = true;
            }
        }

        // 如果从AsyncSemaphore中成功获取信号量，则执行listener这个异步任务
        // 此情景下就是允许该线程向redis发送SUB命令
        if (run) {
            listener.run();
        }
    }

    /**
     * 从AsyncSemaphore的listeners集合中将正在等待的异步任务移除
     * @param listener 异步任务
     * @return listeners中存在这个异步任务且删除成功返回true，否则false
     */
    public boolean remove(Runnable listener) {
        synchronized (this) {
            // 由于Entry已经覆写了equals方法，只要listener相同就是相同的Entry，不在乎permits
            return listeners.remove(new Entry(listener, 0));
        }
    }

    public int getCounter() {
        return counter;
    }

    /**
     * 当一个占有AsyncSemaphore的线程成功执行完毕(此处就是成功向redis发送了SUB命令)，需要释放掉占有的信号量。
     * 该方法总是释放1个信号量。
     */
    public void release() {
        // 当前AsyncSemaphore的等待队列中是否有其他线程在等待
        Entry entryToAcquire = null;

        synchronized (this) {
            // AsyncSemaphore总信号量+1
            counter++;
            Iterator<Entry> iter = listeners.iterator();
            // 遍历AsyncSemaphore.listeners，找到一个需求的permits比当前AsyncSemaphore总信号量小的Entry
            // 在此情境下，由于发送SUB命令所需的信号量都是1，所以就是一个先进先出队列，总之最终所有等待发送SUB命令的线程都能发送成功。
            if (iter.hasNext()) {
                Entry entry = iter.next();
                if (entry.getPermits() <= counter) {
                    // 将其从listeners中删除
                    iter.remove();
                    entryToAcquire = entry;
                }
            }
        }

        // 上面的synchronized找到了一个等待中的线程，OK再执行acquire方法去处理它。
        if (entryToAcquire != null) {
            acquire(entryToAcquire.getRunnable(), entryToAcquire.getPermits());
        }
    }

    @Override
    public String toString() {
        return "value:" + counter + ":queue:" + queueSize();
    }


}
